package suncertify.db;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import suncertify.application.Log;
import suncertify.application.Strings;

/**
 * This class provides a synchronized thread safe cache of record number and lock cookie pairs of values. The values are
 * stored in an internal <code>Map&lt;Long,Long&gt;</code> member variable. The record number is the key of the map
 * element, while the lock cookie is the value of the map element.
 * 
 * <p>
 * Records are locked by calling the {@link #lockRecord(long)} method, which generates a lock cookie for the record and
 * adds the record number and lock cookie to the map. Records are unlocked by calling the
 * {@link #unlockRecord(long, long)} method which removes the record number and lock cookie from the map.
 * 
 * <p>
 * The {@link #lockRecord(long)} and {@link #unlockRecord(long, long)} methods are synchronized by an internal static
 * {@link Lock} object, which guarantees that all calls to lock and unlock are synchronized.
 * 
 * <p>
 * An attempt to lock a record that is already locked, causes the executing thread to give up CPU cycles and wait until
 * another thread signals to all waiting threads that a record is unlocked. Calling the unlock method removes the lock
 * of the specified record from the internal map and signals all blocked threads to wake up, the JVM decides which
 * waiting threads is waked up.
 * 
 * The class should be initialized once in the application, it is the responsibility of the caller to guarantee this.
 * 
 */
class LockManager {

    /**
     * This key/value map contains all locked records and their corresponding cookies. The record number is the map
     * element's key, the lock cookie is the element's value.
     */
    private Map<Long, Long> lockedRecords = new HashMap<Long, Long>();

    /**
     * Provides synchronization access to the record/cookie map. See {@link Lock}
     */
    private static Lock lock = new ReentrantLock();

    /**
     * Provides a wait and notify thread mechanism. See {@link Condition}
     */
    private static Condition lockCondition = lock.newCondition();

    /**
     * Defines the server state. True means threads can serialize and block on the lock while false means a
     * stop/terminate server event was triggered. If this variable is false then all threads should thre
     * IllegalStateException.
     * 
     */
    private boolean serverIsRunning = true;

    /**
     * Constructs the class. Sets the {@link #serverIsRunning} class variable to true.
     */
    public LockManager() {

	this.serverIsRunning = true;
    }

    /**
     * Locks the record provided by <code>recordNo</code> and generates a corresponding lock cookie.
     * <ul>
     * <li>If the internal {@link Lock} is already acquired by another thread then the current thread becomes disabled
     * for thread scheduling purposes and lies dormant until the lock has been acquired.</li>
     * <li>After the executing thread acquires the lock, it examines if the record number exists in the internal
     * record/cookie map, if yes then the current thread gives up the CPU and waits until the record is unlocked.</li>
     * <li>If the record is not locked then a lock cookie is generated and the record number and cookie are added to the
     * internal record/cookie map, then the internal lock is released and the function returns.</li>
     * </ul>
     * 
     * <p>
     * Note:
     * <ul>
     * <li>The function is thread safe.</li>
     * <li>If the provided record number is already locked then the function does not return immediately, the caller of
     * the function may wait for seconds or minutes or even more!</li>
     * <li>The lock cookie is a simple unique number generated as follows:
     * <code>long lockCooke = System.nanoTime();</code></li>
     * </ul>
     * 
     * @param recordNo
     *            The record number to be locked.
     * 
     * @return The lock cookie.
     * 
     * @throws IllegalStateException
     *             Thrown if the {@link #serverIsRunning} class variable is set to false, this is done in the
     *             {@link #releaseAll()} method. In this case the server was stopped. All threads the wake up from the
     *             wait and all threads that have acquired the lock successfully raise this exeption and exit
     *             immediately.
     * 
     */
    public long lockRecord(long recordNo) {

	/**
	 * Use the lock to serialize the access of the multiple threads on the shared lockedRecords Map.
	 */
	lock.lock();

	try {
	    if (!this.serverIsRunning) {
		/**
		 * If the thread returns from the lock then another thread might have set the server to stop state in
		 * this case the thread should exit immediately.
		 */
		throw new IllegalStateException(Strings.MSG_SERVER_NOT_RUNNING);
	    }

	    while (lockedRecords.containsKey(recordNo)) {

		Log.logInfo("LockManager.lockRecord() is going to wait on record " + Long.toString(recordNo));

		/*
		 * If recordNo is in the map then it was already locked by another thread in this case simply wait until
		 * the locking thread releases the locked record by calling signalAll.
		 */
		lockCondition.awaitUninterruptibly();

		/*
		 * Your thread has just woke up!
		 */
		Log.logInfo("LockManager.lockRecord() just returned from awaitUninterruptibly() on record "
			+ Long.toString(recordNo));

		if (!this.serverIsRunning) {
		    /*
		     * Check if another thread stopped the server while your thread was waiting for its record.
		     */
		    throw new IllegalStateException(Strings.MSG_SERVER_NOT_RUNNING);
		}
	    }

	    /**
	     * Congratulations you have made it! you own the lock all others must wait for you to finish and the
	     * recordNo is not locked! So be fast create the cookie for this record and put the pair in the map and
	     * release the lock in the finally clause.
	     */
	    long lockCookie = System.nanoTime();
	    lockedRecords.put(recordNo, lockCookie);

	    Log.logInfo("LockManager.lockRecord() successfully locked record " + Long.toString(recordNo));

	    return lockCookie;
	} finally {
	    lock.unlock();
	}
    }

    /**
     * Unlocks the <code>recordNo</code> using the provided <code>cookie</code> by removing the record number and cookie
     * from the internal map and signaling this removal to all waiting threads by calling the
     * {@link Condition#signalAll()}.
     * 
     * <p>
     * The function is thread safe and synchronized by the internal synchronization {@link Lock}.
     * 
     * @param recordNo
     *            The record number to be unlocked.
     * @param cookie
     *            The lock cookie that was generated for the <code>recordNo</code> in the {@link #lockRecord(long)}
     *            function.
     * @throws SecurityException
     *             Thrown if the <code>cookie</code> parameter is not equal to the cookie in the class record/cookie
     *             map.
     * @throws RecordNotLockedUncheckedException
     *             Thrown if <code>recordNo</code> does not exist in the class' record/cookie map.
     * 
     * @throws IllegalStateException
     *             Thrown if the {@link #serverIsRunning} class variable is set to false, this is done in the
     *             {@link #releaseAll()} method. In this case the server was either stopped or terminated.
     * 
     */
    public void unlockRecord(long recordNo, long cookie) throws SecurityException {

	/**
	 * The lock serializes the access to the share map which can be accessed by many threads at the same time.
	 */
	lock.lock();

	// congratulations you own the lock!
	try {
	    if (!this.serverIsRunning) {
		/**
		 * If the thread returns from the lock then another thread might have set the server to stop state in
		 * this case the thread should exit immediately.
		 */
		throw new IllegalStateException(Strings.MSG_SERVER_NOT_RUNNING);
	    }

	    // check if the record exists in the map
	    if (lockedRecords.containsKey(recordNo)) {

		long lockCookie = lockedRecords.get(recordNo);

		if (lockCookie == cookie) {

		    // remove the record/cookie pair and signal this too all blocking threads.
		    lockedRecords.remove(recordNo);
		    lockCondition.signalAll();

		    Log.logInfo("LockManager.unlockRecord() successfully unlocked record " + Long.toString(recordNo));
		} else {
		    Log.logInfo(Strings.getMessage(Strings.MSG_RECORD_WRONG_UNLOCK_COOKIE, Long.toString(recordNo)));
		    throw new SecurityException(Strings.getMessage(Strings.MSG_RECORD_WRONG_UNLOCK_COOKIE, Long
			    .toString(recordNo)));
		}
	    } else {
		Log.logInfo(Strings.getMessage(Strings.MSG_RECORD_UNLOCK_ERROR, Long.toString(recordNo)));
		throw new RecordNotLockedUncheckedException(Strings.getMessage(Strings.MSG_RECORD_UNLOCK_ERROR, Long
			.toString(recordNo)));
	    }
	} finally {
	    lock.unlock();
	}
    }

    /**
     * Determines if the <code>recNo</code> and the <code>lockCookie</code> exist in the class' record/cookie cache. If
     * the record does not exist in the cache or has a different cookie then the exceptions below are thrown.
     * 
     * <p>
     * This function is not thread safe, it does not use the class' {@link Lock}, this class is only called from
     * synchronized functions.
     * 
     * @param recNo
     *            The record number.
     * @param lockCookie
     *            The lock cookie.
     * @throws SecurityException
     *             Thrown if the record number exists in the record/cookie cache and the cached cookie is different form
     *             the <code>lockCookie</code>.
     * @throws RecordNotLockedUncheckedException
     *             Thrown if the <code>recNo</code> does not exist in the record/cookie cache.
     */
    void isCookieValid(long recNo, long lockCookie) throws SecurityException {

	if (lockedRecords.containsKey(recNo)) {

	    long originalLockCookie = lockedRecords.get(recNo);

	    if (originalLockCookie == lockCookie) {

		return;
	    } else {

		throw new SecurityException(Strings.getMessage(Strings.MSG_RECORD_WRONG_LOCK_COOKIE, Long
			.toString(recNo)));
	    }
	} else {
	    throw new RecordNotLockedUncheckedException(Strings.getMessage(Strings.MSG_RECORD_MODFIY_UNLOCKED, Long
		    .toString(recNo)));
	}
    }

    /**
     * Called when the server is stopped. This method synchronizes on the {@link #lock} and sets the
     * {@link #serverIsRunning} class variable to false and calls class lockCondition.signalAll() method which wakes up
     * all blocking threads. All threads that wake up will throw an {@link IllegalStateException} and exit immediately.
     */
    void releaseAll() {

	lock.lock();

	try {
	    this.serverIsRunning = false;

	    if (this.lockedRecords.size() > 0) {

		lockCondition.signalAll();
	    }
	} finally {
	    lock.unlock();
	}
    }
}
